package com.sinozo.data.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.app.function.MyTimestampsAndWatermarks;
import com.sinozo.data.bean.key.PageViewKeyByBean;
import com.sinozo.data.bean.TrafficPageViewBean;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.*;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * 页面浏览日志
 */
public class DwsTrafficPageViewWindow {

    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(DwsTrafficPageViewWindow.class);
        try {
            //1.获取执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);

            //1.1 开启CheckPoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, org.apache.flink.api.common.time.Time.of(3L, TimeUnit.DAYS), org.apache.flink.api.common.time.Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath, "dwsTrafficPageViewWindow")));
            env.setStateBackend(new EmbeddedRocksDBStateBackend());

            // 2.读取三个主题的数据创建流
            String uvTopic = ConfigConstant.KAFKA_UNIQUE_VISITOR_TOPIC;

            String topic = ConfigConstant.KAFKA_PAGE_LOG_TOPIC;
            String groupId = ConfigConstant.KAFKA_PAGE_VIEW_GROUP_ID;
            DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
            DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

            // 3.统一数据格式
            SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUvDS = uvDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");

                TrafficPageViewBean bean = TrafficPageViewBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).event(page.getString("event"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .subEvent(page.getString("sub_event")).code(common.getString("code"))
                        .uv(1L).pv(0L).duringTime(0L).isNew(common.getLong("is_new")).ts(page.getLong("ts")).build();
                return bean;
            });

            SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithPageDS = pageDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");

                TrafficPageViewBean bean = TrafficPageViewBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).event(page.getString("event"))  .labCode(common.getString("lab_code"))
                        .labGroupCode(common.getString("lab_group_code")).subEvent(page.getString("sub_event")).code(common.getString("code"))
                        .duringTime(page.getLong("during_time")).uv(0L).pv(1L).isNew(common.getLong("is_new")).ts(page.getLong("ts")).build();
                return bean;
            });

            // 4.将流进行Union
            DataStream<TrafficPageViewBean> unionDS = trafficPageViewWithPageDS.union(trafficPageViewWithUvDS);

            // 5.提取事件时间生成WaterMark
            SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithWmDS = unionDS.assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks());

            // 6.分组开窗聚合
            WindowedStream<TrafficPageViewBean, PageViewKeyByBean, TimeWindow> windowedStream = trafficPageViewWithWmDS.keyBy(new KeySelector<TrafficPageViewBean, PageViewKeyByBean>() {
                @Override
                public PageViewKeyByBean getKey(TrafficPageViewBean value) throws Exception {
                    PageViewKeyByBean keyByBean = new PageViewKeyByBean();
                    BeanUtils.copyProperties(keyByBean, value);
                    return keyByBean;
                }
            }).window(TumblingEventTimeWindows.of(DataConfig.WINDOW_TIME));


            SingleOutputStreamOperator<TrafficPageViewBean> resultDS = windowedStream.reduce(new ReduceFunction<TrafficPageViewBean>() {
                @Override
                public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                    value1.setPv(value1.getPv() + value2.getPv());
                    value1.setUv(value1.getUv() + value2.getUv());
                    value1.setDuringTime(value1.getDuringTime() + value2.getDuringTime());
                    return value1;
                }
            }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, PageViewKeyByBean, TimeWindow>() {
                @Override
                public void apply(PageViewKeyByBean key, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                    //获取数据
                    TrafficPageViewBean next = input.iterator().next();
                    //补充信息
                    next.setWinStartTime(DateFormatUtil.toYmdHms(window.getStart()));
                    next.setWinEndTime(DateFormatUtil.toYmdHms(window.getEnd()));
                    //修改TS
                    next.setTs(System.currentTimeMillis());
                    next.setCreateTime(DateFormatUtil.toYmdHms(next.getTs()));
                    //输出数据
                    out.collect(next);
                }
            });

            // 7.将数据写出或更新到ClickHouse
            resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_traffic_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

            // 8.启动任务
            env.execute("DwsTrafficPageViewWindow");
        } catch (Exception e) {
            logger.error("error exec", e);
        }

    }


}
